package com.wfh.job;


import com.google.common.collect.Sets;
import com.wfh.config.RabbitMQConfig;
import com.wfh.constant.ThumbConstant;
import com.wfh.listener.thumb.msg.ThumbEvent;
import com.wfh.model.entity.Thumb;
import com.wfh.service.ThumbService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author FengHuan Wang
 * @Date 2025/4/20 10:49
 * @Version 1.0
 */
@Service
@Slf4j
public class ThumbReconcileJob {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private ThumbService thumbService;

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 自定义线程池
     */
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2,
            4,
            60,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(4), Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.CallerRunsPolicy());

    /**
     * 定时任务，每天凌晨2点执行
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void run(){
        long startTime  = System.currentTimeMillis();
        // 获取该时间分片下的所有用户id
        Set<Long> userIds = new HashSet<>();
        String pattern = ThumbConstant.USER_THUMB_KEY_PREFIX + "*";
        try (Cursor<String> cursor = redisTemplate.scan(ScanOptions.scanOptions().match(pattern).count(1000).build())) {
            while (cursor.hasNext()){
                String next = cursor.next();
                Long userId = Long.valueOf(next.replace(ThumbConstant.USER_THUMB_KEY_PREFIX, ""));
                userIds.add(userId);
            }
        }

        // 使用线程池并行处理比对任务
        threadPoolExecutor.execute(() -> {
            // 逐个用户比对
            userIds.forEach(userId -> {
                Set<Long> redisBlogIds = redisTemplate.opsForHash().keys(ThumbConstant.USER_THUMB_KEY_PREFIX + userId).stream().map(obj -> Long.valueOf(obj.toString())).collect(Collectors.toSet());
                Set<Long> mysqlBlogIds = Optional.ofNullable(thumbService.lambdaQuery().eq(Thumb::getUserId, userId).list()).orElse(new ArrayList<>()).stream().map(Thumb::getBlogId).collect(Collectors.toSet());
                // 计算差异
                Sets.SetView<Long> difference = Sets.difference(redisBlogIds, mysqlBlogIds);
                // 发送补偿事件
                sendEvents(userId, difference);
            });
        });
        log.info("补偿任务完成，耗时{}ms", System.currentTimeMillis() - startTime);
    }

    /**
     * 发送补偿事件到消息队列
     * @param userId
     * @param blogIds
     */
    private void sendEvents(Long userId, Set<Long> blogIds){
        blogIds.forEach(blogId -> {
            ThumbEvent thumbEvent = new ThumbEvent(userId, blogId, ThumbEvent.EventType.INCR, LocalDateTime.now());
            try {
                rabbitTemplate.convertAndSend(RabbitMQConfig.THUMB_EXCHANGE, RabbitMQConfig.THUMB_ROUTING_KEY, thumbEvent);
            } catch (Exception ex) {
                log.error("补偿事件发送失败：userId={}, blogId={}, ex={}", userId, blogId, ex.getMessage());
            }
        });
    }
}
